package com.chenfan.mcn.mq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.chenfan.ccp.plug.log.util.TraceIdUtil;
import com.chenfan.ccp.util.start.ApplicationContextUtil;
import com.chenfan.mcn.clients.finance.FinanceClient;
import com.chenfan.mcn.config.UserVoContextHolder;
import com.chenfan.mcn.constant.DingTalkTemplateCst;
import com.chenfan.mcn.extension.queue.SmartDelayQueue;
import com.chenfan.mcn.component.DingTalkAlarmSender;
import com.chenfan.mcn.mq.message.AbstractMetaMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * @author 2381
 * 生产者
 */

@Slf4j
public abstract class AbstractMessageProducer {

    @Autowired
    private RocketMQTemplate rocketTemplate;

    @Value("${topic.mcn_to_finance:data_to_finance}")
    protected String financeTopic;

    @Autowired
    private DingTalkAlarmSender alarmSender;

    @Autowired
    private FinanceClient financeClient;

    final protected <T extends AbstractMetaMessage> boolean doSendRecordToMq(String businessCode, T data, String springTopic) {
        return this.doSendRecordToMq(businessCode, data, springTopic, null);
    }

    final protected <T extends AbstractMetaMessage> boolean doSendRecordToMq(String businessCode, T data, String springTopic, String hashKey) {
        log.info("MQ消息发送同步【租户id={}】", UserVoContextHolder.getTenantId());
        //生成message类型
        String dataJsonString = JSONObject.toJSONString(data, SerializerFeature.WriteClassName);
        String traceId = TraceIdUtil.getTraceId();
        Message<byte[]> message = MessageBuilder.withPayload(dataJsonString.getBytes(StandardCharsets.UTF_8))
                .setHeader("KEYS", businessCode)
                .setHeader(TraceIdUtil.SHORT_TRACE_ID, traceId)
                .setHeader("tenantId", UserVoContextHolder.getTenantId())
                .setHeader("userId", UserVoContextHolder.getUserId())
                .setHeader("token", UserVoContextHolder.getToken())
                .build();
        String messageName = data != null ? data.getClass().getSimpleName() : "message";
        log.info("mq start sending {}：business_code={}, topic={}, body={} ,message={}", messageName, businessCode, springTopic, dataJsonString, JSON.toJSONString(message));
        SendResult sendResult = null;
        if (StringUtils.isBlank(hashKey)) {
            sendResult = rocketTemplate.syncSend(springTopic, message);
        } else {
            sendResult = rocketTemplate.syncSendOrderly(springTopic, message, hashKey);
        }
        if(Objects.nonNull(sendResult) && SendStatus.SEND_OK.equals(sendResult.getSendStatus())) {
            log.info("mq send sucess：business_code={}, result={}", businessCode, sendResult);
            return true;
        } else {
            log.error("mq send error：business_code={}, result={}", businessCode, sendResult);
            alarmSender.sendMarkdown("费用消息发送失败", String.format(DingTalkTemplateCst.MQ_FAIL, ApplicationContextUtil.getProfile(), ApplicationContextUtil.getServiceName(),
                    businessCode, UserVoContextHolder.getDefaultUserName(),traceId, alarmSender.getSkipUrl(traceId), JSONObject.toJSONString(sendResult))
             );
            return false;
        }
    }

    /**
     * 检查费用生成情况
     * @param chargeSourceCode
     */
    protected void checkCharge(String chargeSourceCode) {
        String traceId = TraceIdUtil.getTraceId();
        SmartDelayQueue.post(() -> {
            Integer count = financeClient.getInvalid(chargeSourceCode, null, null).getObj();
            if(ObjectUtils.defaultIfNull(count, 0).intValue() == 0) {
                alarmSender.sendMarkdown("5分钟内费用未生成", String.format(DingTalkTemplateCst.CHARGE_CREATE_FAIL, ApplicationContextUtil.getProfile(), ApplicationContextUtil.getServiceName(),
                        chargeSourceCode, UserVoContextHolder.getDefaultUserName(), traceId, alarmSender.getSkipUrl(traceId)));
            }
        }, 5, TimeUnit.MINUTES);
    }
}
